package com.millstein;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class JoinTest5 {

    public static void main(String[] args) {
        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        tableEnv.executeSql(
                "create table t1 (\n" +
                "    id string, \n" +
                "    pt as proctime() \n" +
                ") with (\n" +
                "    'connector' = 'kafka',\n" +
                "    'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "    'properties.group.id' = 'tsing',\n" +
                "    'topic' = 's1',\n" +
                "    'format' = 'csv'\n" +
                ")"
        );

        tableEnv.executeSql(
                "create table dic (\n" +
                "    dic_code string,\n" +
                "    dic_name string\n" +
                ") with (\n" +
                "    'connector' = 'jdbc',\n" +
                "    'url' = 'jdbc:mysql://hadoop102:3306/gmall?useSSL=false',\n" +
                "    'table-name' = 'base_dic',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456'\n" +
                ")"
        );

        tableEnv
                .sqlQuery(
                        "select t1.id, d.dic_name from t1 join dic for system_time as of t1.pt as d on t1.id = d.dic_code"
        )
                .execute()
                .print();
    }
}
